
#include "rpcserver.h"
#include <google/protobuf/descriptor.h>
#include "../proto/rpcheader.pb.h"
#include "../proto/rpcresponse.pb.h"
#include "../redis/redisclient.h"
#include "tinyrpc/base/timestamp.h"
#include "tinyrpc/base/logger/logger.h"
#include "tinyrpc/base/logger/logger.h"
#include "tinyrpc/net/coroutine_function.h"
namespace TinyRPC{

void TinyRPCServer::init(const std::string &filePath){
    m_config.init(filePath);
    std::string redis_ip = m_config.LookUp<std::string>("redis.ip");
    int redis_port = m_config.LookUp<int>("redis.port");
    m_register.init(redis_ip, redis_port);

    int listen_port = m_config.LookUp<int>("server.port");
    std::string server_ip = m_config.LookUp<std::string>("server.ip");
    m_serverAddress = INetAddress(server_ip, listen_port);
}


void TinyRPCServer::registService(google::protobuf::Service * service){
    
    const google::protobuf::ServiceDescriptor * serviceDescrip = service->GetDescriptor();
    std::string serviceName = serviceDescrip->name();

    int methodCount = serviceDescrip->method_count();
    ServiceInfo serviceInfo;
    for(int i=0;i<methodCount; i++){
        
        serviceInfo.m_service = service;
        const google::protobuf::MethodDescriptor * methodDescrip = serviceDescrip->method(i);
        serviceInfo.m_methodMap.emplace(methodDescrip->name(), methodDescrip);   
    }
    m_serviceMap.insert({serviceName, serviceInfo});

}


// 处理一次连接就是不停的读取内容
// 因为一个connection 可能会发生多个 rpc request
void TinyRPCServer::co_handleRPCConnection(SocketChannelPtr socketChannel){
    LOG_INFO<<"in co_handleRPCConnection";
    while (true){
        // 当其长时间不发送时， 如何让协程主动停止

        int len = socketChannel->receiveFromRemote();
        if(len == 0) {
            socketChannel->close();
            LOG_INFO<<"socketChannel close"; 
            break;
        }
        // 重新设置事件
        
        m_conns[socketChannel->getFd()]->setLastRequestTime(TimeStamp::Now());
        
        // 只要解码器能够解码到一个完整的消息就起一个协程来处理消息
        std::string rpc_str = socketChannel->read(m_codec);
        while (rpc_str.size()>0)
        {
            LOG_INFO<<rpc_str;
            EventLoop::CurrentEventLoop()->runInLoop(Task(std::bind(&TinyRPCServer::co_handleOneRequest, this, socketChannel, rpc_str), true));
            rpc_str = socketChannel->read(m_codec);
        }
        
    }
}

// 处理一个客户端连接里面的一次RPC请求
void TinyRPCServer::co_handleOneRequest(SocketChannelPtr socketChannel, std::string rpc_str){
    /**  
     *      rpc header
     *              msg_no
     *              service_name
     *              method_name
     *              arg;
     */

    
    m_conns[socketChannel->getFd()]->increaseRequestTime();
    
    RPCHeader rpc_header;
   
    
    // 处理messageHeader
    rpc_header.ParseFromString(rpc_str);
    
    // 寻找对于的reponse
    auto it_service = m_serviceMap.find(rpc_header.service_name());
    auto it_method = it_service->second.m_methodMap.find(rpc_header.method_name());
    
    google::protobuf::Service * service = it_service->second.m_service;
    const google::protobuf::MethodDescriptor * method = it_method->second;

    google::protobuf::Message * request = service->GetRequestPrototype(method).New();
    request->ParseFromString(rpc_header.args());

    google::protobuf::Message * response = service->GetResponsePrototype(method).New();
    // 给下面的mthod方法的调用
    uint32_t msg_no = rpc_header.msg_no();
    
    //google::protobuf::Closure * closure = google::protobuf::NewCallback<TinyRPCServer, SocketChannelPtr, ClosureParam>(this, &TinyRPCServer::sendRpcResponse, socketChannel, param);
    try{
        service->CallMethod(method, nullptr, request, response, nullptr);
    }catch(...){
        sendRpcResponse(socketChannel,  msg_no, 0, response);
        delete request;
        delete response;
        return;
    }
    sendRpcResponse(socketChannel,  msg_no, 1, response);
     delete request;
    delete response;
    // 寻找
    // socketChannel.write(str);
    // std::cout<<str<<std::endl;
}

// 当应用程序调用完之后所使用发送程序
void TinyRPCServer::sendRpcResponse(SocketChannelPtr socketChannel, uint32_t msg_no, int code, google::protobuf::Message * response){
    
    RPCResponse rpcResponse; 
    std::string send_str;
    rpcResponse.set_msg_no(msg_no);
    if(code == 0){
        rpcResponse.set_code(0);
        rpcResponse.SerializeToString(&send_str);
        socketChannel->write(send_str, m_codec);
        m_conns[socketChannel->getFd()]->increaseResponseTime();
        return ;
    }
    

    rpcResponse.set_code(1);
    std::string response_str;
    response->SerializeToString(&response_str);
    rpcResponse.set_response(response_str);
    rpcResponse.SerializeToString(&send_str);
    socketChannel->write(send_str, m_codec);
    m_conns[socketChannel->getFd()]->increaseResponseTime();
}


void TinyRPCServer::co_accept(){
    
    ServerSocketChannel server = ServerSocketChannel::CreateAndBind(m_serverAddress);
    server.listen();
    while (1)
    {
        SocketChannelPtr socketChannel = server.accept();
        // loop->runInLoop(Task(std::bind(&TcpServer::echo, this, socketChannel), true));
        std::cout<<"accept socket"<<std::endl;
        // 这里会存在问题， 即在一个协程里面调用另一个协程
        EventLoop * loop = getNextLoop();
        RPCConnectionPtr conn = std::make_shared<RPCConnection>(socketChannel, loop, TimeStamp::Now());
        {
            LockGuard<CoLock> guard(m_connLock);
            m_conns[socketChannel->getFd()] = conn;

        }
        
        loop->runInLoop(Task(std::bind(&TinyRPCServer::co_handleRPCConnection, this, socketChannel), true));
        socketChannel.reset();
    }
}

TinyRPCServer::~TinyRPCServer(){
    m_register.delService(m_serviceMap, m_serverAddress);
}

void TinyRPCServer::start(){
   
    m_eventLoopPool.start();

    getNextLoop()->runInLoop(Task(std::bind(&TinyRPCServer::co_accept, this), true));

    getNextLoop()->queueInLoop(Task(std::bind(&TinyRPCServer::co_clearInActiveConn, this), true));

    int keyExpiredTime =  m_config.LookUp<int>("redis.keyExpiredTime");
    int heartbeatTime = m_config.LookUp<int>("redis.heartbeatTime");
     // 向注册中心上报
    m_register.regist(m_serviceMap, m_serverAddress,  keyExpiredTime);
    m_register.publish("ADDSERVICE", "add");
    LOG_INFO<<"redis register finish";
    // 加入心跳机制
    getNextLoop()->runInLoop(Task(std::bind(&RedisRegister::co_heartbeat, &(this->m_register), heartbeatTime, keyExpiredTime), true));

    
    m_baseLoopPtr->loop();

    // 

}

void TinyRPCServer::co_clearInActiveConn(){
    coroutine_sleep(3);
    while (!m_quit)
    {
        LOG_INFO<<"co_clearInActiveConn";
        TimeStamp now = TimeStamp::Now();
        {
            LockGuard<CoLock> guard(m_connLock);
            for(auto it = m_conns.begin(); it != m_conns.end();){
                // 如果大于一秒没有连接
                if(now - it->second->getLastRequestTime() > TimeStamp(TimeStamp::kMicroSecPerSecond) && !it->second->haveNotHandledRequest()){
                    //it->second->getConn()->deleteEvent(it->second->getEventLoop(), WriteEvent | ReadEvent);
                    it->second->getConn()->close();
                    it = m_conns.erase(it);
                }else{
                    it++;
                }
            } 
        }
        
        coroutine_sleep(3);
    }
    
     
}
}



